package com.fjec.im.rabbitmq.consumer;

import com.fjec.common.Const;
import com.fjec.im.service.IBizSystemMessageService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

@Slf4j
@Component
public class SysMsgConsumer {

    @Resource
    private IBizSystemMessageService systemMessageService;

    @Resource(name = "compensateAmqpTemplate")
    private RabbitTemplate compensateAmqpTemplate;

    @Value("${fjec.compensate.complete.router.key}")
    private String compensateRouteKey;

    @Value("${fjec.compensate.complete.exchange.name}")
    private String compensateExchange;

    /**
     * 消费系统消息
     *
     * @param message
     * @param channel
     * @param headers
     * @throws Exception
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(
                            value = "${fjec.market.sysmsg.queue.name}",
                            durable = "${fjec.market.sysmsg.queue.durable}"
                    ),
                    exchange = @Exchange(
                            value = "${fjec.market.sysmsg.exchange.name}",
                            durable = "${fjec.market.sysmsg.exchange.durable}",
                            type = "${fjec.market.sysmsg.exchange.type}",
                            ignoreDeclarationExceptions = "${fjec.market.sysmsg.exchange.ignoreDeclarationExceptions}"
                    ),
                    key = "${fjec.market.sysmsg.router.key}"
            )
    )
    @RabbitHandler
    public void onSysMsg(Message message, Channel channel, @Headers Map<String, Object> headers) throws Exception {
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            systemMessageService.handleMessage(message);
            channel.basicAck(deliveryTag, true);
        } catch (DuplicateKeyException dex) {
            channel.basicAck(deliveryTag, true);
        } catch (Exception ex) {
            log.error(ex.getMessage(), ex);
            channel.basicReject(deliveryTag, false);
        }
    }

    /**
     * 发送补偿确认消息
     *
     * @param dataId
     * @param msgType
     * @param message
     */
    private void sendCompensateMsg(Long dataId, Integer msgType, Message message) {
        MessageProperties props = new MessageProperties();
        props.setHeader(Const.MQ_HEADER_DATA_ID_KEY, dataId);
        props.setHeader(Const.MQ_HEADER_MSG_TYPE_KEY, msgType);
        Message sendMsg = new Message(message.getBody(), props);
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(msgType + "||" + dataId);
        compensateAmqpTemplate.convertAndSend(compensateExchange, compensateRouteKey, sendMsg, correlationData);

    }
}
